package com.shujia.core

import com.shujia.core.Demo5WaterMark.Event1
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.api.scala._

object Demo5WaterMark {
  def main(args: Array[String]): Unit = {

    /**
      * 1,1574666000
      * 1,1574671000
      * 1,1574667000
      * 1,1574668000
      * 1,1574669000
      * 1,1574670000
      * 1,1574671000
      * 1,1574672000
      * 1,1574673000
      * 1,1574674000
      * 1,1574675000
      * 1,1574680000
      * 1,1574760000
      *
      */

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //当所有线程中时间到达窗口的结束时间才开始计算
    env.setParallelism(1)
    //处理时间 TimeCharacteristic.ProcessingTime
    //TimeCharacteristic.EventTime   时间时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val ds = env.socketTextStream("node1", 8888)

    val eventDS = ds.map(line => {
      val split = line.split(",")

      Event1(split(0), split(1).toLong)
    })

    //统计最近5秒每个车出现的次数

    eventDS

      .assignTimestampsAndWatermarks(new MyAssignerWithPeriodicWatermarks)

      .map(event => (event.id, 1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))
      .reduce((x, y) => (x._1, x._2 + y._2))
      .print()

    env.execute("Demo4EventTime")


  }

  case class Event1(id: String, time: Long)

}

/**
  * 自定义时间字段  和水位线
  *
  */

class MyAssignerWithPeriodicWatermarks extends AssignerWithPeriodicWatermarks[Event1] {
  //所有数据里面最大的时间
  var maxTime: Long = _


  //获取当前watermark
  //getCurrentWatermark  定时运行
  override def getCurrentWatermark: Watermark = {

    println("getCurrentWatermark")

    //最大的时间减去5     最大允许数据延迟的时间
    new Watermark(maxTime - 5000)

  }

  //获取时间字段
  override def extractTimestamp(element: Event1, previousElementTimestamp: Long): Long = {
    println("extractTimestamp")
    if (element.time > maxTime) {
      maxTime = element.time
    }
    element.time
  }
}